package com.atliuzu.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atliuzu.utils.DimUtil;
import com.atliuzu.utils.DruidDSUtil;
import com.atliuzu.utils.ThreadPoolUtil;
import lombok.SneakyThrows;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.util.Collections;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @Author Administrator
 * @Time 2022-08-21-22:50
 */
public abstract class  DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements AsyncJoinFunction<T>{

    private ThreadPoolExecutor threadPoolExecutor=null;
    private DruidDataSource druidDataSource=null;


    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    public DimAsyncFunction() {

    }

    @Override
    public void open(Configuration parameters) throws Exception {
        threadPoolExecutor= ThreadPoolUtil.getThreadPoolExecutor();
        druidDataSource= DruidDSUtil.createDataSource();
    }

    @Override
    public void asyncInvoke(T input, ResultFuture<T> resultFuture) {
        threadPoolExecutor.execute(new Runnable() {
            @SneakyThrows
            @Override
            public void run() {

                //建立连接
                DruidPooledConnection connection = druidDataSource.getConnection();

                //查询维表
                String key=getKey(input);
                JSONObject dimInfo = DimUtil.getDimInfo(connection, tableName, key);

                //补充信息
                if (dimInfo!=null){
                    join(input,dimInfo);
                }

                //关闭连接
                connection.close();

                //将数据写出
                resultFuture.complete(Collections.singletonList(input));
            }
        });
    }

    @Override
    public void timeout(T input, ResultFuture<T> resultFuture) throws Exception {
        System.out.println("TimeOut>>>>>>>>>>>" + input);
    }
}